package com.bw.yk01;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import scala.Tuple7;

import java.util.Properties;



public class FlinkTM5_2 {
    public static void main(String[] args) throws Exception {
        EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
        TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);


        /*
        create table t_yk8(
            order_id string,
            create_time string,
            baseProvinceid string,
            baseProvincename string,
            skuInfoid string,
            skuInfoname string,
             skuInfonum int
        )with(
           'connector' = 'filesystem',
           'path' = 'datas',
           'format' = 'csv'
        )
         */
        bbTableEnv.executeSql("create table t_yk8(\n" +
                "            order_id string,\n" +
                "            create_time string,\n" +
                "            baseProvinceid string,\n" +
                "            baseProvincename string,\n" +
                "            skuInfoid string,\n" +
                "            skuInfoname string,\n" +
                "            userId string,\n" +
                "             skuInfonum int\n" +
                "        )with(\n" +
                "           'connector' = 'filesystem',\n" +
                "           'path' = 'datas',\n" +
                "           'format' = 'csv'\n" +
                "        )");

        //bbTableEnv.sqlQuery("select * from t_yk8").execute().print();

        //2）、按城市统计热销top5商品，热销规则为该商品销售量在该城市占比超过12%；（4分）
        /*
        select * from (
        select *,row_number() over(partition by baseProvinceid order by skusums desc) rk from (
            select *,sum(skusums) over(partition by baseProvinceid) provincesums from (
         select baseProvinceid,skuInfoid,sum(skuInfonum) skusums from t_yk8 group by baseProvinceid,skuInfoid)
         ) where cast(skusums as double)/provincesums >=0.12 ) where rk<=5
         */
        /*bbTableEnv.sqlQuery("select * from (\n" +
                        "        select *,row_number() over(partition by baseProvinceid order by skusums desc) rk from (\n" +
                        "            select *,sum(skusums) over(partition by baseProvinceid) provincesums from (\n" +
                        "         select baseProvinceid,skuInfoid,sum(skuInfonum) skusums from t_yk8 group by baseProvinceid,skuInfoid)\n" +
                        "         ) where cast(skusums as double)/provincesums >=0.12 ) where rk<=5")
                .execute().print();
*/


        //3）、按地区统计不同品类商品的渗透率，渗透率=复购量/首次购买量；（4分）
        //TODO 没理解公式意思，按照复购率求 复购率 = 购买2次以上的人数 / 购买总人数
        /*
        select baseProvinceid,skuInfoid,userId,count(1) n from t_yk8 group by baseProvinceid,skuInfoid,userId
        select baseProvinceid,skuInfoid,count(1) f1 from t1 where n >=2 group by baseProvinceid,skuInfoid
         */

        //求每个省份，每个商品，每个人购买次数
        Table t1 = bbTableEnv.sqlQuery("select baseProvinceid,skuInfoid,userId,count(1) n from t_yk8 group by baseProvinceid,skuInfoid,userId");
        bbTableEnv.createTemporaryView("t1",t1);

        //求购买2次以上的人数
        Table t2 = bbTableEnv.sqlQuery("select baseProvinceid,skuInfoid,count(1) f1 from t1 where n >=2 group by baseProvinceid,skuInfoid");
        bbTableEnv.createTemporaryView("t2",t2);

        //购买总人数
        Table t3 = bbTableEnv.sqlQuery("select baseProvinceid,skuInfoid,count(1) f2 from t1 group by baseProvinceid,skuInfoid");
        bbTableEnv.createTemporaryView("t3",t3);



        bbTableEnv.sqlQuery("select t2.baseProvinceid,t2.skuInfoid, cast(f1 as double)/f2 from t2 join t3" +
                " on t2.baseProvinceid = t3.baseProvinceid "+
                " and t2.skuInfoid = t3.skuInfoid ").execute().print();




    }
}
